package com.tmsb.taimu_main;

import com.alibaba.fastjson.JSON;
import com.tmsb.pojo.MysqlBinlog;
import com.tmsb.pojo.TaimuMainUserStockBuy;
import com.tmsb.pojo.TaimuMainUserTiXian;
import com.tmsb.sink.UserStockBuySink;
import com.tmsb.sink.UserTiXianSink;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.Const;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Description: 用taimu_main的user_stock_buy来获取折扣兑换绩效积分数量和折扣两个实时指标
 * Version:1.0.0
 *
 * @author xiekb
 * @date 2020/7/4
 */
public class taimuMainAnalysis {
    private static final OutputTag<TaimuMainUserStockBuy> USER_STOCK_BUY = new OutputTag<TaimuMainUserStockBuy>("user_stock_buy") {};
    private static final OutputTag<TaimuMainUserTiXian> USER_TIXIAN = new OutputTag<TaimuMainUserTiXian>("user_tixian"){};
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //  env.setStateBackend(new FsStateBackend("file:///C:\\Users\\home\\Desktop\\Taimu_main\\Tixian"));
      env.setStateBackend(new FsStateBackend(ConfUtil.getUserStockBuyCheckpointUri()));
    //    env.setStateBackend(new FsStateBackend(ConfUtil.getUserTiXianCheckpointUri()));
        // 取消作业时保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.enableCheckpointing(ConfUtil.getCheckpointInterval(), CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(ConfUtil.getCheckpointMinPauseBetween());
        env.setParallelism(ConfUtil.getParallelism());

       // 重启策略
       env.setRestartStrategy(RestartStrategies.failureRateRestart(
                ConfUtil.getfailureRate(), org.apache.flink.api.common.time.Time.of(ConfUtil.getfailureInterval(), TimeUnit.MINUTES),
                org.apache.flink.api.common.time.Time.of(ConfUtil.getdelayInterval(), TimeUnit.SECONDS)
        ));


        // source

        DataStream<String> source = env
                .addSource(new FlinkKafkaConsumer<>(ConfUtil.getTopicOfTaiMuMain(), new SimpleStringSchema(), ConfUtil.getKafkaProperties()))
                .uid("kafka-source");

        SingleOutputStreamOperator<Object> process =
                source
                .map(s -> JSON.parseObject(s, MysqlBinlog.class)).uid("map-parser-pojo")
                .process(new ProcessFunction<MysqlBinlog, Object>() {

                    @Override
                    public void processElement(MysqlBinlog mysqlBinlog, Context context,
                                               Collector<Object> collector) throws Exception {
                        if (Const.MysqlDatabase.TAIMU_MAIN.toString().equalsIgnoreCase(mysqlBinlog.getDatabase())) {
                            if (Const.MysqlTable.USER_STOCK_BUY.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                                TaimuMainUserStockBuy userStockBuy = JSON.parseObject(mysqlBinlog.getData(), TaimuMainUserStockBuy.class);
                                if (userStockBuy == null) {
                                    return;
                                }
                                context.output(USER_STOCK_BUY, userStockBuy);
                            }
                            else if(Const.MysqlTable.USER_TIXIAN.toString().equalsIgnoreCase(mysqlBinlog.getTable())) {
                                TaimuMainUserTiXian userTiXian = JSON.parseObject(mysqlBinlog.getData(), TaimuMainUserTiXian.class);
                                if (userTiXian == null) {
                                    return;
                                }
                                context.output(USER_TIXIAN, userTiXian);
                            }
                        }
                    }
                }).uid("process-side-output");

        process.getSideOutput(USER_STOCK_BUY)
                .addSink(new UserStockBuySink())
                .uid("score-sink");

  //      process.getSideOutput(USER_TIXIAN).print("User_Tixian");
        process.getSideOutput(USER_TIXIAN)
         .addSink(new UserTiXianSink())
       .uid("Tixian-sink");

        try {
            env.execute("taimuMainAnalysis");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
